package window;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.junit.Test;

import java.util.Iterator;

public class Flink_fullWin {
    public static class MyWondow implements WindowFunction<Tuple2<String, Integer>,Tuple2<String, Integer>,Tuple,TimeWindow>{
        @Override
        public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Integer>> input, Collector<Tuple2<String, Integer>> out) throws Exception {
            Iterator<Tuple2<String, Integer>> iterator = input.iterator();

            //获取key值
            String key = tuple.getField(0);

            int count=0;
            //计算当前数据条数
            while (iterator.hasNext()){
                Tuple2<String, Integer> next = iterator.next();
                count = next.f1 + count;

            }

            //输出数据
            out.collect(new Tuple2<>(key,count));

        }
    }
    //TODO 自定义窗口计算函数
    @Test
    public void test() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        env.socketTextStream("node193",9999)
                .flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                        for (String s : value.split(" ")) {
                            out.collect(new Tuple2<>(s,1));
                        }
                    }
                })
                .keyBy(value -> value.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .apply(new WindowFunction<Tuple2<String, Integer>,Tuple2<String, Integer>,String,TimeWindow>() {
                    @Override
                    public void apply(String key, TimeWindow window, Iterable<Tuple2<String, Integer>> input, Collector<Tuple2<String, Integer>> out) throws Exception {
                        Iterator<Tuple2<String, Integer>> iterator = input.iterator();
                        int sum=0;
                        while (iterator.hasNext()) {
                            sum = iterator.next().f1 + sum;
                        }
                        out.collect(new Tuple2<>(key,sum));
                    }
                }).print();
        env.execute();
    }
}
